package com.spring.batch.base.config;

import com.spring.batch.base.JobCompletionNotificationListener;
import com.spring.batch.base.Person;
import com.spring.batch.base.PersonItemProcessor;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * ClassName: SpringBatchConfig
 * Description:
 * Date: 2017/9/21 11:50
 *
 * @author SAM SHO
 * @version V1.0
 */
@Configuration
@EnableBatchProcessing
public class SpringBatchConfig {

    @Autowired
    private DataSource dataSource;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Bean
    @StepScope
    public ItemReader<Person> reader(@Value("#{jobParameters['input.file.name']}") String path) {
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
//        reader.setResource(new ClassPathResource("sample-data.csv"));
        reader.setResource(new ClassPathResource(path));

        /*DefaultLineMapper<Person> lineMapper = new DefaultLineMapper<>();

        // 分隔符
        DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
        lineTokenizer.setDelimiter(",");
        lineTokenizer.setNames(new String[]{"firstName", "lastName"});
        lineMapper.setLineTokenizer(lineTokenizer);

        // 数据转成类
        BeanWrapperFieldSetMapper<Person> fieldSetMapper = new BeanWrapperFieldSetMapper<>();
        fieldSetMapper.setTargetType(Person.class);
        lineMapper.setFieldSetMapper(fieldSetMapper);

        reader.setLineMapper(lineMapper);*/


        // 对cvs 文件的数据和领域模型类做对应映射,同上面的编码
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setDelimiter(DelimitedLineTokenizer.DELIMITER_COMMA);//","分割
                setNames(new String[]{"firstName", "lastName"});
            }});

            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});

        }});

        return reader;
    }

    @Bean
    public ItemProcessor<Person, Person> processor() {
        return new PersonItemProcessor();
    }

    @Bean
    @StepScope
    public ItemWriter<Person> writer(@Value("#{jobParameters['output.insert.sql']}") String sql) {
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setDataSource(dataSource);
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
//        writer.setSql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)");
        writer.setSql(sql);
        return writer;
    }

    @Bean
    public JobRepository jobRepository() throws Exception {
        JobRepositoryFactoryBean jobRepository = new JobRepositoryFactoryBean();
        jobRepository.setDataSource(dataSource);
        jobRepository.setTransactionManager(transactionManager);
        jobRepository.setDatabaseType("mysql");
        return jobRepository.getObject();
    }


    @Bean
    public JobLauncher launcher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository());
        jobLauncher.setTaskExecutor(new SyncTaskExecutor());
        return jobLauncher;
    }


    @Bean
    public Job job() throws Exception {
        JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository());
        return jobBuilderFactory.get("job")
                .incrementer(new RunIdIncrementer())
                .flow(step())
                .end()
                .listener(listener())
                .build();
    }

    @Bean
    public Step step() throws Exception {
        StepBuilderFactory stepBuilderFactory = new StepBuilderFactory(jobRepository(), transactionManager);

        return stepBuilderFactory.get("step")
                .<Person, Person>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .listener(listener())
                .build();
    }


    @Bean
    public JobExecutionListener listener() {
        return new JobCompletionNotificationListener(jdbcTemplate);
    }

}
